package com.iot.service.mqtt.service.pub;

import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.service.StoreStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @Author: hszhou
 * @Date: 2025/2/19 16:54
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.REDIS)
@Slf4j
public class RedisPublishStoreServiceImpl implements IPublishStoreService {


    private final static String WILL_KEY = "will:";

    private final static String PUBLISH_KEY = "publish:";

    private final ReactiveRedisTemplate<String,Object> reactiveRedisTemplate;

    @Override
    public Mono<TlMqttPublishReq> savePublishMessage(String clientId, Long messageId, TlMqttPublishReq req) {
        return reactiveRedisTemplate.opsForHash().put(PUBLISH_KEY+clientId, String.valueOf(messageId), req)
                .flatMap(value->findPublishMessage(clientId,messageId))
                .doOnError(e -> log.error("savePublishMessage failed for key: {}", PUBLISH_KEY+clientId, e));
    }

    @Override
    public Mono<TlMqttPublishReq> clearPublishMessage(String clientId, Long messageId) {

        return reactiveRedisTemplate.opsForHash().remove(PUBLISH_KEY+clientId, String.valueOf(messageId))
                  .flatMap(e->findPublishMessage(clientId,messageId))
                  .doOnSuccess(e->{

                     // log.info("删除了【{}】的消息【{}】",messageId,e);
                  }) .doOnError(e -> log.error("clearPublishMessage failed for key: {}", PUBLISH_KEY+clientId, e));
    }

    @Override
    public Mono<Boolean> clearPublishMessage(String clientId) {
        return reactiveRedisTemplate.opsForHash().delete(PUBLISH_KEY+clientId)
                .doOnError(e -> log.error("clearPublishMessage byClientId failed for key: {}", PUBLISH_KEY+clientId, e));
    }

    @Override
    public Mono<TlMqttPublishReq> findPublishMessage(String clientId, Long messageId) {
        return reactiveRedisTemplate.opsForHash().get(PUBLISH_KEY+clientId, String.valueOf(messageId))
                .doOnError(e -> log.error("findPublishMessage byClientId failed for key: {}", PUBLISH_KEY+clientId, e))
                .cast(TlMqttPublishReq.class);

    }

    @Override
    public Flux<TlMqttPublishReq> findPublishMessage(String clientId) {

        return reactiveRedisTemplate.opsForHash().keys(PUBLISH_KEY+clientId)
                .doOnError(e -> log.error("findPublishMessage byClientId failed for key: {}", PUBLISH_KEY+clientId, e))
                .cast(TlMqttPublishReq.class);
    }

    @Override
    public Mono<Boolean> saveWillMessage(String clientId, TlMqttPublishReq req) {
        return reactiveRedisTemplate.opsForValue().set(WILL_KEY+clientId,req)
                .doOnError(e -> log.error("saveWillMessage byClientId failed for key: {}", WILL_KEY+clientId, e));
    }

    @Override
    public Mono<TlMqttPublishReq> findWillMessage(String clientId) {
        return reactiveRedisTemplate.opsForValue().get(WILL_KEY+clientId)
                .doOnError(e -> log.error("findWillMessage byClientId failed for key: {}", WILL_KEY+clientId, e))
                .cast(TlMqttPublishReq.class);
    }

    @Override
    public Mono<Boolean> clearWillMessage(String clientId) {
        return reactiveRedisTemplate.opsForValue().delete(WILL_KEY+clientId)
                .doOnError(e -> log.error("clearWillMessage byClientId failed for key: {}", WILL_KEY+clientId, e));
    }
}
